package org.example.flink;

import com.google.common.collect.Lists;
import org.apache.calcite.sql.SqlInsert;
import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.parser.SqlParser;
import org.apache.commons.io.FileUtils;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.internal.TableEnvironmentImpl;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.api.java.internal.StreamTableEnvironmentImpl;
import org.apache.flink.table.delegation.Parser;
import org.apache.flink.table.delegation.Planner;
import org.apache.flink.table.operations.CatalogSinkModifyOperation;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.ddl.CreateTableOperation;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.util.List;

/**
 * @author mcclone
 */
public class SQLFileExec {
    private static final Logger logger = LoggerFactory.getLogger(SQLFileExec.class);

    public static void main(String[] args) {
        try {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            EnvironmentSettings settings = EnvironmentSettings.newInstance()
                    .useBlinkPlanner()
                    .inStreamingMode()
                    .build();
            env.enableCheckpointing(5000);
            StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);
//tEnv.registerFunction("",);
            String sql = FileUtils.readFileToString(new File("D:\\code\\flink-demo\\src\\main\\resources\\flink.sql"));
            for (String sqlItem : sql.split(";")) {
                if (tEnv instanceof TableEnvironmentImpl) {
                    Planner planner = ((StreamTableEnvironmentImpl) tEnv).getPlanner();
                    Parser parser = planner.getParser();
                    List<Operation> operations = parser.parse(sqlItem);
                    Operation operation = operations.get(0);
                    if (operation instanceof CreateTableOperation) {
                        tEnv.sqlUpdate(sqlItem);
                    } else if (operation instanceof CatalogSinkModifyOperation) {
//                        tEnv.sqlUpdate(sqlItem);
                        //以下为统计所需
                        SqlParser.Config config = SqlParser.configBuilder().build();
                        SqlParser parserCa = SqlParser.create(sqlItem, config);
                        SqlNode node = parserCa.parseQuery();
                        if (node instanceof SqlInsert) {
                            SqlNode source = ((SqlInsert) node).getSource();
                            if (source.getKind().belongsTo(Lists.newArrayList(SqlKind.SELECT))) {
                                String selectSQL = source.toString();
                                Table table = tEnv.sqlQuery(selectSQL);
                                tEnv.toAppendStream(table, Row.class)
                                        .timeWindowAll(Time.minutes(1))
                                        .process(new ProcessAllWindowFunction<Row, Row, TimeWindow>() {
                                            @Override
                                            public void process(Context context, Iterable<Row> elements, Collector<Row> out) throws Exception {
                                                for (Row element : elements) {
                                                    out.collect(element);
                                                }
                                                System.out.println(elements.spliterator().estimateSize());
                                            }
                                        })
                                        .print();
                            }
                        }
                    }
                }

            }
            env.execute("SQLFileExec");
        } catch (Exception e) {
            e.printStackTrace();
            logger.error(e.getMessage(), e);
            throw new RuntimeException(e);
        }
    }
}
